﻿using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Linq.Expressions;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;

namespace RabbitBus
{
    public class RabbitMQBus : BaseBus, IBus
    {


        private const string _exchangeFail = "OverallExchange.Fail";      //定义一个用于接收 失败 消息的全局交换机
        private const string _queueFail = "OverallQueue.Fail";            //定义一个用于接收 失败 消息的全局队列
        private bool _delayedMsg = false; //是否延时队列

        /// <summary>
        /// 存放消费者对应的队列名称
        /// 使用线程安全字典存放，防止多线程问题
        /// </summary>
        public static ConcurrentDictionary<string, (Type, object)> consumerQueueDictionary = new ConcurrentDictionary<string, (Type, object)>();

        /// <summary>
        /// MQ初始化
        /// </summary>
        public RabbitMQBus(bool delayedMsg = false)
        {
            _delayedMsg = delayedMsg;
            RabbitMQInit(ConsumeMessages);
        }

        /// <summary>
        /// 消息队列初始化
        /// </summary>
        /// <param name="ConsumeAction">消费者委托，可跟据场景传入指定消费者</param>
        private void RabbitMQInit(Action<IConnection, IModel, string> ConsumeAction)
        {
            ConnectionFactory connectionfactory = CreateConnectionFactory.CreateRabbitMqConnectionFactory();
            IConnection connection = connectionfactory.CreateConnection();
            IModel channel = connection.CreateModel();
            System.Reflection.Assembly[] allAssembly = AppDomain.CurrentDomain.GetAssemblies();
            //获取当前类所在程序集
            System.Reflection.Assembly.GetExecutingAssembly();
            List<Type> consumers = new List<Type>();
            Assembly.GetAssembly(typeof(ConsumerBase<>));
            foreach (Assembly assembly in allAssembly)
            {
                var types = assembly.GetTypes().Where(x => (x?.BaseType?.IsGenericType ?? false) && x.GetInterfaces().Count() > 0).ToList();
                consumers.AddRange(types);
            }
            foreach (var consumer in consumers)
            {
                var genericArguments = consumer.BaseType.GetGenericArguments();
                if (genericArguments == null || !genericArguments.Any())
                    continue;
                var attributions = consumer.GetCustomAttributes(typeof(QueueConsumerAttribute), false);
                if (attributions == null || attributions.Length == 0)
                    continue;
                var attribution = attributions[0] as QueueConsumerAttribute;
                if (attribution == null || string.IsNullOrWhiteSpace(attribution.QueueName))
                    continue;
                var queueName = attribution.QueueName;
                var exchange = attribution.QueueName;
                var exchangeType = attribution.ExchangeType;
                switch (exchangeType)
                {
                    case ExchangeType.Fanout:
                        FanoutQueueInit(connection, channel, queueName, exchange);
                        break;
                    case ExchangeTypeExpand.Delayed:
                        DelayedQueueInit(connection, channel, queueName, exchange);
                        break;
                    default:
                        break;
                }
                ConsumeAction.Invoke(connection, channel, queueName);
                if (!consumerQueueDictionary.ContainsKey(queueName))
                {
                    object consutomInstance = Activator.CreateInstance(consumer);
                    consumerQueueDictionary.TryAdd(queueName, (consumer, consutomInstance));
                }
            }
        }


        /// <summary>
        /// 消息发布
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="msg"></param>
        public void Publish<T>(T msg, int time = 0) where T : BusEvent
        {
            try
            {
                ConnectionFactory factory = CreateConnectionFactory.CreateRabbitMqConnectionFactory();
                using IConnection connection = factory.CreateConnection();
                using IModel channel = connection.CreateModel();
                {
                    var queueName = string.Empty;
                    var exchange = string.Empty;
                    foreach (var consumer in consumerQueueDictionary)
                    {
                        var genericArguments = consumer.Value.Item1.BaseType.GetGenericArguments();
                        if (typeof(T) != genericArguments.Single())
                            continue;
                        msg.Queue = consumer.Key;
                        exchange = consumer.Key;
                        string sendMsg = JsonConvert.SerializeObject(msg);
                        MsgPublish(connection, channel, sendMsg, exchange, time);
                    }
                }
            }
            catch (Exception e)
            {
                Console.WriteLine(e.Message);
            }
        }

        /// <summary>
        /// 消息发布
        /// </summary>
        /// <param name="connection"></param>
        /// <param name="channel"></param>
        /// <param name="msg"></param>
        /// <param name="exchange"></param>
        private void MsgPublish(IConnection connection, IModel channel, string msg, string exchange, int time = 0)
        {
            var body = Encoding.UTF8.GetBytes(msg);
            IBasicProperties basicProperties = channel.CreateBasicProperties();
            basicProperties.Persistent = true;
            basicProperties.DeliveryMode = 2;
            basicProperties.ContentEncoding = "UTF-8";
            if (_delayedMsg)
                basicProperties.Headers = new Dictionary<string, object>() { { "x-delay", time } };
            channel.ConfirmSelect();//开启发送消息确认
            channel.BasicPublish(exchange: exchange,
                    routingKey: string.Empty, //Fnout模式不用填写routingKey
                    basicProperties: basicProperties,
                    body: body);
            var sendState = channel.WaitForConfirms();//如果消息被rabbitMQ接收到，则为true
            if (!sendState)
            {
                Console.WriteLine("消息发送失败，需要重试");
            }
        }





        /// <summary>
        ///  Fanout队列消息初始化
        /// </summary>
        /// <param name="connection"></param>
        /// <param name="channel"></param>
        /// <param name="queueName">队列名称</param>
        /// <param name="exchange"></param>
        private void FanoutQueueInit(IConnection connection, IModel channel, string queueName, string exchange)
        {
            var retryQueueName = $"{queueName}.Retry";
            var retryExchange = $"{queueName}.Retry";

            //设置prefetchCount : 1来告知RabbitMQ，在未收到消费端的消息确认时，不再分发消息，也就确保了当消费端处于忙碌状态时
            channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

            //声明交换机
            channel.ExchangeDeclare(exchange: exchange, type: ExchangeType.Fanout, durable: true, autoDelete: false, arguments: null);
            //声明重试交换机
            channel.ExchangeDeclare(exchange: retryExchange, type: ExchangeType.Fanout, durable: true, autoDelete: false, arguments: null);
            //声明全局错误交换机
            channel.ExchangeDeclare(exchange: _exchangeFail, type: ExchangeType.Fanout, durable: true, autoDelete: false, arguments: null);

            //定义队列参数
            var queueNormalArgs = new Dictionary<string, object>();
            {
                queueNormalArgs.Add("x-dead-letter-exchange", _exchangeFail);   //指定死信交换机，用于将 Noraml 队列中失败的消息投递给 Fail 交换机
            }
            var queueRetryArgs = new Dictionary<string, object>();
            {
                queueRetryArgs.Add("x-dead-letter-exchange", exchange);  //指定死信交换机，用于将 Retry 队列中超时的消息投递给 Noraml 交换机
                queueRetryArgs.Add("x-message-ttl", 60000);                     //定义 queueRetry 的消息最大停留时间 (原理是：等消息超时后由 broker 自动投递给当前绑定的死信交换机)                                                               //定义最大停留时间为防止一些 待重新投递 的消息、没有定义重试时间而导致内存溢出
            }
            var queueFailArgs = new Dictionary<string, object>();
            {
                //暂无
            }
            if (_delayedMsg)
            {
                queueNormalArgs.Add("x-delayed-type", ExchangeType.Fanout);
                queueRetryArgs.Add("x-delayed-type", ExchangeType.Fanout);
                queueFailArgs.Add("x-delayed-type", ExchangeType.Fanout);
            }

            //声明队列信息
            channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: queueNormalArgs);
            //声明重试队列信息
            channel.QueueDeclare(queue: retryQueueName, durable: true, exclusive: false, autoDelete: false, arguments: queueRetryArgs);
            //声明全局错误队列
            channel.QueueDeclare(queue: _queueFail, durable: true, exclusive: false, autoDelete: false, arguments: queueFailArgs);

            //队列信息绑定
            channel.QueueBind(queue: queueName, exchange: exchange, routingKey: "");
            //重试队列信息绑定
            channel.QueueBind(queue: retryQueueName, exchange: retryExchange, routingKey: "");
            //全局错误队列信息绑定
            channel.QueueBind(queue: _queueFail, exchange: _exchangeFail, routingKey: "");
        }


        /// <summary>
        ///  Fanout队列消息初始化
        /// </summary>
        /// <param name="connection"></param>
        /// <param name="channel"></param>
        /// <param name="queueName">队列名称</param>
        /// <param name="exchange"></param>
        private void DelayedQueueInit(IConnection connection, IModel channel, string queueName, string exchange)
        {
            var retryQueueName = $"{queueName}.Retry";
            var retryExchange = $"{queueName}.Retry";

            //设置prefetchCount : 1来告知RabbitMQ，在未收到消费端的消息确认时，不再分发消息，也就确保了当消费端处于忙碌状态时
            channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
            //设置Exchange队列类型
            var argMaps = new Dictionary<string, object>() { { "x-delayed-type", ExchangeType.Fanout } };

            //声明交换机
            channel.ExchangeDeclare(exchange: exchange, type: ExchangeTypeExpand.Delayed, durable: true, autoDelete: false, arguments: argMaps);
            //声明重试交换机
            channel.ExchangeDeclare(exchange: retryExchange, type: ExchangeTypeExpand.Delayed, durable: true, autoDelete: false, arguments: argMaps);
            //声明全局错误交换机
            channel.ExchangeDeclare(exchange: _exchangeFail, type: ExchangeType.Fanout, durable: true, autoDelete: false, arguments: null);

            //定义队列参数
            var queueNormalArgs = new Dictionary<string, object>();
            {
                queueNormalArgs.Add("x-dead-letter-exchange", _exchangeFail);   //指定死信交换机，用于将 Noraml 队列中失败的消息投递给 Fail 交换机
            }
            var queueRetryArgs = new Dictionary<string, object>();
            {
                queueRetryArgs.Add("x-dead-letter-exchange", exchange);  //指定死信交换机，用于将 Retry 队列中超时的消息投递给 Noraml 交换机
                queueRetryArgs.Add("x-message-ttl", 60000);                     //定义 queueRetry 的消息最大停留时间 (原理是：等消息超时后由 broker 自动投递给当前绑定的死信交换机)                                                               //定义最大停留时间为防止一些 待重新投递 的消息、没有定义重试时间而导致内存溢出
            }
            var queueFailArgs = new Dictionary<string, object>();
            {
                //暂无
            }
            if (_delayedMsg)
            {
                queueNormalArgs.Add("x-delayed-type", ExchangeType.Fanout);
                queueRetryArgs.Add("x-delayed-type", ExchangeType.Fanout);
                queueFailArgs.Add("x-delayed-type", ExchangeType.Fanout);
            }

            //声明队列信息
            channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: queueNormalArgs);
            //声明重试队列信息
            channel.QueueDeclare(queue: retryQueueName, durable: true, exclusive: false, autoDelete: false, arguments: queueRetryArgs);
            //声明全局错误队列
            channel.QueueDeclare(queue: _queueFail, durable: true, exclusive: false, autoDelete: false, arguments: queueFailArgs);

            //队列信息绑定
            channel.QueueBind(queue: queueName, exchange: exchange, routingKey: "");
            //重试队列信息绑定
            channel.QueueBind(queue: retryQueueName, exchange: retryExchange, routingKey: "");
            //全局错误队列信息绑定
            channel.QueueBind(queue: _queueFail, exchange: _exchangeFail, routingKey: "");
        }


        /// <summary>
        /// 消费消息
        /// </summary>
        private void ConsumeMessages(IConnection connection, IModel channel, string queueName)
        {
            #region 创建一个普通消息消费者
            {
                try
                {
                    var consumerEvent = new EventingBasicConsumer(channel);
                    channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumerEvent);
                    consumerEvent.Received += (sender, e) =>
                    {
                        try
                        {
                            var _sender = (EventingBasicConsumer)sender;            //消息传送者
                            var _channel = _sender.Model;                           //消息传送通道
                            var _message = (BasicDeliverEventArgs)e;                //消息传送参数
                            var _headers = _message.BasicProperties.Headers;        //消息头
                            var _content = Encoding.UTF8.GetString(_message.Body.ToArray());  //消息内容
                            var _death = default(Dictionary<string, object>);       //死信参数
                            if (_headers != null && _headers.ContainsKey("x-death"))
                                _death = (Dictionary<string, object>)(_headers["x-death"] as List<object>)[0];
                            if (string.IsNullOrWhiteSpace(_content))
                                return;
                            var contentEvent = JsonConvert.DeserializeObject<BusEvent>(_content);
                            var queueName = contentEvent.Queue;
                            try
                            #region 消息处理
                            {
                                if (_death != null)
                                {
                                    //获取死信队列请求头时间
                                    //AmqpTimestamp amqpTimestamp = (AmqpTimestamp)_death["time"];
                                    //long num = amqpTimestamp.UnixTime * 10000000;
                                    //long ticks = new DateTime(1970, 1, 1, 8, 0, 0).Ticks + num;
                                    //DateTime dateTime = new DateTime(ticks);
                                }
                                var body = e.Body;
                                var message = Encoding.UTF8.GetString(body.ToArray());
                                Console.WriteLine();
                                Console.WriteLine($"{DateTime.Now.ToString("HH:mm:ss.fff")}\t(1.0)消息接收：\r\n\t[deliveryTag={_message.DeliveryTag}]\r\n\t[consumerID={_message.ConsumerTag}]\r\n\t[exchange={_message.Exchange}]\r\n\t[routingKey={_message.RoutingKey}]\r\n\t[content={_content}]");
                                if (message.Contains("异常"))
                                    throw new Exception("模拟消息处理失败效果。");
                                #region 消息执行
                                if (consumerQueueDictionary.ContainsKey(queueName))
                                {
                                    var genericArguments = consumerQueueDictionary[queueName].Item1.BaseType.GetGenericArguments();
                                    var consutomContent = JsonConvert.DeserializeObject(_content, genericArguments.Single());
                                    var type = consumerQueueDictionary[queueName].Item1;

                                    //dynamic consutomInstance = Activator.CreateInstance(type);
                                    //consutomInstance.Consume(consutomContent);
                                    //var method = consumerQueueDictionary[queueName].Item1.GetMethod("Consume");
                                    //if (method != null)
                                    //    method.Invoke(consumerQueueDictionary[queueName].Item2, new object[] { consutomContent });
                                    ConfigureHandler(genericArguments.Single(), consumerQueueDictionary[queueName].Item1, queueName, _content);
                                }
                                #endregion
                                //处理成功时
                                Console.WriteLine($"{DateTime.Now.ToString("HH:mm:ss.fff")}\t(1.1)处理成功：\r\n\t[deliveryTag={_message.DeliveryTag}]");
                                //消息确认 (销毁当前消息)
                                _channel.BasicAck(deliveryTag: _message.DeliveryTag, multiple: false);
                            }
                            #endregion
                            catch (Exception ex)
                            {
                                #region 消息处理失败时
                                var retryCount = (long)(_death?["count"] ?? default(long)); //查询当前消息被重新投递的次数 (首次则为0)
                                Console.WriteLine($"{DateTime.Now.ToString("HH:mm:ss.fff")}\t(1.2)处理失败：\r\n\t[deliveryTag={_message.DeliveryTag}]\r\n\t[retryCount={retryCount}]");

                                if (retryCount >= 2)
                                #region 投递第3次还没消费成功时，就转发给 exchangeFail 交换机
                                {
                                    //消息拒绝（投递给死信交换机，也就是上边定义的 ("x-dead-letter-exchange", _exchangeFail)）
                                    _channel.BasicNack(deliveryTag: _message.DeliveryTag, multiple: false, requeue: false);
                                    Console.WriteLine($"消息进入死信队列{_queueFail}");
                                }
                                #endregion
                                else
                                #region 否则转发给 exchangeRetry 交换机
                                {
                                    var interval = (retryCount + 1) * 10; //定义下一次投递的间隔时间 (单位：秒)
                                                                          //如：首次重试间隔10秒、第二次间隔20秒、第三次间隔30秒
                                                                          //定义下一次投递的间隔时间 (单位：毫秒)
                                    _message.BasicProperties.Expiration = (retryCount * 1000).ToString();
                                    //消息确认 (销毁当前消息)
                                    _channel.BasicAck(deliveryTag: _message.DeliveryTag, multiple: false);
                                    //将消息投递给 _exchangeRetry (会自动增加 death 次数)
                                    var retryExchange = $"{queueName}.Retry";
                                    _channel.BasicPublish(exchange: retryExchange, routingKey: _message.RoutingKey, basicProperties: _message.BasicProperties, body: _message.Body);
                                }
                                #endregion
                                #endregion
                            }
                        }
                        catch (Exception ex)
                        {

                            // TODO：当发生全局异常，则记录代码日志
                            Console.WriteLine(ex.Message);
                        }
                    };

                }
                catch (Exception e)
                {
                    Console.WriteLine(e.Message);
                }
            }
            #endregion

        }


        /// <summary>
        /// 配置事件处理器
        /// </summary>
        /// <param name="eventType">事件类型</param>
        /// <param name="handlerType">事件处理器类型，类型必须符合IEventHandler{TEvent}</param>
        public void ConfigureHandler(Type eventType, Type handlerType, string queueName, string concurrent)
        {
            this.FastInvoke(new Type[] { eventType, handlerType },
                x => x.ConsumerTo<BusEvent, IEventConsumer<BusEvent>>(queueName, concurrent),
                new object[] { queueName, concurrent });
        }

        /// <summary>
        /// 处理
        /// </summary>
        /// <typeparam name="TEvent"></typeparam>
        /// <typeparam name="TConsumer"></typeparam>
        /// <param name="_content"></param>
        public void ConsumerTo<TEvent, TConsumer>(string queueName, string _content)
         where TConsumer : IEventConsumer<TEvent>
         where TEvent : BusEvent
        {
            TEvent consutomContent = JsonConvert.DeserializeObject<TEvent>(_content);
            TConsumer consutomInstance = (TConsumer)consumerQueueDictionary[queueName].Item2;
            consutomInstance.Consume(consutomContent);
        }
    }

}
